package com.ultrapower.request;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Scan;

import com.sematext.hbase.wd.RowKeyDistributorByHashPrefix;
import com.ultrapower.mapper.ResponseTimeMapper;
import com.ultrapower.util.ApplicationMapStatisticsUtils;
import com.ultrapower.util.HBaseTables;
import com.ultrapower.util.TimeSlot;
import com.ultrapower.util.rowkey.RangeOneByteSimpleHash;
import com.ultrapower.util.topology.ResponseTime;
/**
 * 
 * @author wangy
 *
 */
public class ApplicationMapStatisticsSelf_Ver2Table extends BaseRequest {

	private TimeSlot timeSlot = new TimeSlot();
	private int scanCacheSize = 40;
	private RowKeyDistributorByHashPrefix rowKeyDistributorByHashPrefix = new RowKeyDistributorByHashPrefix(new RangeOneByteSimpleHash(0, 32, 8));
	
	public static void main(String[] args) {
		BaseRequest table=new ApplicationMapStatisticsSelf_Ver2Table();
		table.execute();
	}

	@Override
	void execute() {
		TableName mapStatisticsSelfTableName = this.getTableName(HBaseTables.MAP_STATISTICS_SELF_VER2_STR);
		this.getHbaseTemplate().find(mapStatisticsSelfTableName, createScan(), rowKeyDistributorByHashPrefix,
				resultScanner -> {
					ResponseTimeMapper map = new ResponseTimeMapper(rowKeyDistributorByHashPrefix);
					resultScanner.forEach(c -> {
						try {
							ResponseTime r = map.mapRow(c, 0);
							System.out.println(r);
						} catch (Exception e) {
							e.printStackTrace();
						}
					});
					return null;
				});
	}

	@Override
	protected Scan createScan() {
		final long startTime = timeSlot.getTimeSlot(from) - 1;
		final long endTime = timeSlot.getTimeSlot(to);
		// start key is replaced by end key because timestamp has been reversed
		byte[] startKey = ApplicationMapStatisticsUtils.makeRowKey(this.applicationName, this.srviceTypeCode, endTime);
		byte[] endKey = ApplicationMapStatisticsUtils.makeRowKey(this.applicationName, this.srviceTypeCode, startTime);
		final Scan scan = new Scan();
		scan.setCaching(this.scanCacheSize);
		scan.setStartRow(startKey);
		scan.setStopRow(endKey);
		scan.addFamily(HBaseTables.MAP_STATISTICS_SELF_VER2_CF_COUNTER);
		scan.setId("ApplicationSelfScan");

		return scan;
	}

}
